package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

public class ExSlidingWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExSlidingWindows demo = new ExSlidingWindows();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = demo.exSlidingWindows(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }


    /**
     * 滑动窗口具有固定大小，并按指定的滑动间隔滑动。【如果滑动间隔小于窗口大小，则滑动窗口重叠。】
     * 因此，可以将行分配给多个窗口。例如，一个 15 分钟大小和 5 分钟滑动间隔的滑动窗口将每行分配给 3 个 15 分钟大小的不同窗口，这些窗口以 5 分钟的间隔进行评估。滑动窗口可以按事件时间、处理时间或行计数进行定义。
     *
     * @param orders
     * @return
     */
    private Table exSlidingWindows(Table orders) {
        Table result1 = orders.window(Slide.over(lit(30).minutes())
                .every(lit(10).minutes())
                .on($("rowtime"))
                .as("w"))
                .groupBy($("a"), $("w")) // group by key and window
                .select(
                        $("a"),
                        $("w").start(),
                        $("w").end(),
                        $("w").rowtime(),
                        $("b").count()
                );
        //.window(Slide.over(rowInterval(10)).every(rowInterval(5)).on($("proctime")).as("w"));TODO
        return result1;
//        1> +I[liji, 2024-01-24T18:40, 2024-01-24T19:10, 2024-01-24T19:09:59.999, 2]
//        1> +I[liji, 2024-01-24T18:50, 2024-01-24T19:20, 2024-01-24T19:19:59.999, 3]
//        1> +I[liji, 2024-01-24T19:00, 2024-01-24T19:30, 2024-01-24T19:29:59.999, 3]
    }


}
